package com.bbx.flink.table_api_or_sql.table_api_all;

import com.bbx.flink.demo.entity.Person;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Over;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.Tumble;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.*;


/**
 * Scan, Projection, and Filter
 * Column Operations
 * Aggregations
 * https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/tableApi.html#scan-projection-and-filter
 */
public class Demo1 {

    private static StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    private static StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


    public static void main(String[] args) throws Exception {
        //切换实验不同方法
        int flag = 9;
        env.setParallelism(1);

        String createSql = "CREATE TABLE outputTable (\n" +
                " id String,name String,tel String " +
                ") WITH (\n" +
                "  'connector.type' = 'filesystem',\n" +
                "  'connector.path' = 'file:///C:\\Users\\kaifacs\\Desktop\\33.csv', \n" +
                "  'format.type' = 'csv' \n" +
                ")";

        //创建表
        tableEnv.executeSql(createSql);

        switch (flag) {
            case 1: {
                //Operations   Scan, Projection, and Filter~~~~~~~~~~  from()~~~~~~~~~~~~~~~~~~~~~~~~~~
                from();
                break;
            }
            case 2: {
                //Operations  Scan, Projection, and Filter~~~~~~~~~~   fromValues()/where()~~~~~~~~~~~~~~~~~~~
                fromValues();
                break;
            }
            case 3: {
                //Operations  Scan, Projection, and Filter~~~~~~~~~~    fromValues()~~~~~指定属性类型~~~~~~~~~~~~~~~~~~
                fromValuesByRow();
                break;
            }
            case 4: {
                //Opreations Scan, Projection, and Filter~~~~~~~~~~   AS()/filter()~~~~~~~~~~~~~~~~~~~~~~~~~
                as();
                break;
            }
            case 5: {
                //Opreations Column Operations   ~~~~~~~~~~  AddColumns() /DropColumns()/AddOrReplaceColumns()/RenameColumns()~~~~~~~~~~~~~~
                columnOperations();
                break;
            }
            case 6: {
                //Opreations Aggregations   ~~~~~~~~~~ GroupBy  ~~~~~~~~~~~~~~
                aggregationsGroupBy();
                break;
            }
            case 7: {
                //Opreations Aggregations   ~~~~~~~~~~ distinct ~~~~~~~~~~~~~~
                opreationsDistinct();
                break;
            }
            case 8: {
                //Opreations Aggregations   ~~~~~~~~~~ GroupBy Window ~~~~~~~~~~~~~~
                groupByWindow();
                break;
            }
            case 9: {
                //Opreations Aggregations   ~~~~~~~~~~ Over Window ~~~~~~~~~~~~~~
                overWindow();
                break;
            }
        }


        env.execute();

    }

    /**
     * //Operations   ~~~~~~~~~~~~~~~~~~~~~~~~~  from()~~~~~~~~~~~~~~~~~~~~~~~~~~
     */
    static void from() {
        Table fromTable = tableEnv.from("outputTable");
        //打印输出表字段
        fromTable.printSchema();
        Table select = fromTable.select($("id"), $("name"), $("tel"));
        tableEnv.toAppendStream(select, Row.class).print("select");
    }


    /**
     * Operations ~~~~~~~~~~~~~~~~~fromValues()~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
     */
    static void fromValues() {
        Table fromValuesTable = tableEnv.fromValues(
                row(1, "a", 1.22),
                row(2, "b", 2.22),
                row(3, "c", 3.22),
                row(4, "d", 4.22),
                row(5, "e", 5.22)
        );
        fromValuesTable.printSchema();
        Table select1 = fromValuesTable.select($("f0"), $("f1"), $("f2"))
                .where($("f0").isGreater(3));
        tableEnv.toAppendStream(select1, Row.class).print("fromValues");
    }

    /**
     * //Operations ~~~~~~~~~~~~~~~~~fromValues()~~~~~指定属性类型~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
     */
    static void fromValuesByRow() {
        Table fromValuesTable = tableEnv.fromValues(
                DataTypes.ROW(
                        DataTypes.FIELD("id", DataTypes.INT()),
                        DataTypes.FIELD("name", DataTypes.VARCHAR(5)),
                        DataTypes.FIELD("val", DataTypes.DECIMAL(3, 2))
                ),
                row(1, "a", 1.22),
                row(2, "b", 2.22),
                row(3, "c", 3.22),
                row(4, "d", 4.22),
                row(5, "e", 5.22)
        );
        fromValuesTable.printSchema();
        Table select1 = fromValuesTable.select($("id"), $("name"), $("val"))
                .where($("id").isGreater(3));
        tableEnv.toAppendStream(select1, Row.class).print("fromValues");
    }

    static void as() {
        Table as = tableEnv.from("outputTable").as("as_id", "as_name", "as_tel");
        tableEnv.toAppendStream(
                as.select($("*"))
                        .filter($("as_id")
                                .isLessOrEqual("10")),
                Row.class)
                .print("AS~~~~");
    }

    /**
     * Column Operations
     */
    static void columnOperations() {
        Table outputTable = tableEnv.from("outputTable");
        //addColumns() 相当于给查询结果增加一个属性,可以是基于现有字段，也可以是固定内容
        Table addColumnsTable = outputTable.addColumns(concat("022-", $("tel")).as("addColumn"))
                .addColumns(concat("aaaa", "bbbb").as("addColumnAAAA"));
        addColumnsTable.printSchema();
        tableEnv.toAppendStream(addColumnsTable, Row.class).print("addcolumn~~~~~~");
        //AddOrReplaceColumns()
        Table addReplaceColumnsTable = addColumnsTable.addOrReplaceColumns(concat("022--", $("tel").as("addColumn")))
                .addOrReplaceColumns(concat("add", "Replace", "column").as("addReplaceColumn"));
        tableEnv.toAppendStream(addReplaceColumnsTable, Row.class).print("addReplaceColumns~~~~~~");
        addReplaceColumnsTable.printSchema();
        //dropColumns()
        Table dropColumnsTable = addReplaceColumnsTable.dropColumns($("addReplaceColumn"));
        tableEnv.toAppendStream(dropColumnsTable, Row.class).print("dropColumnsTable");
        dropColumnsTable.printSchema();
        //RenameColumns                                                   addColumn
        Table renameColumnTable = dropColumnsTable.renameColumns($("addColumn").as("renameColumn"));
        tableEnv.toAppendStream(renameColumnTable, Row.class).print("renameColumnTable");
    }

    /**
     * aggregations    GroupBy()
     */
    static void aggregationsGroupBy() {
        Table aggregationsGroupByTable = tableEnv.from("outputTable")
                .groupBy($("id"), $("name"), $("tel"))
                .select($("id"), $("name"), $("tel"), $("id").count().as("countId"));

        tableEnv.toRetractStream(aggregationsGroupByTable, Row.class).print("aggregationsGroupByTable");
    }

    /**
     * Similar to a SQL DISTINCT clause. Returns records with distinct value combinations.
     */
    static void opreationsDistinct() {
        Table distinctTable = tableEnv.from("outputTable").distinct();
        tableEnv.toRetractStream(distinctTable, Row.class).print("distinctTable");
    }

    /**
     * Groups and aggregates a table on a group window and possibly one or more grouping keys.
     * processing time  window
     * 例：输入
     * 37389,孙克静,84813992
     * 37390,霍建春,24685366
     * 37389,孙克静,84813992
     * 37389,孙克静,84813992
     * 37389,孙克静,84813992
     * 输出
     * windowTable> (true,37389,孙克静,84813992,4,2021-01-28T01:50,2021-01-28T01:52,2021-01-28T01:52:00.004)
     * windowTable> (true,37390,霍建春,24685366,1,2021-01-28T01:50,2021-01-28T01:52,2021-01-28T01:52:00.005)
     */
    static void groupByWindow() {
        DataStream<Person> dataStream = env.socketTextStream("192.168.10.131", 10003)
                .map(i -> {
                    String[] split = i.split(",");
                    return new Person(split[0], split[1], split[2]);
                });

        Table windowTable = tableEnv.fromDataStream(dataStream,
                $("id"), $("name"), $("phone"), $("proctime").proctime())
                .window(Tumble.over(lit(2).minutes()).on($("proctime")).as("aaaa"))
                .groupBy($("id"), $("name"), $("phone"), $("aaaa"))
                .select($("id"),
                        $("name"),
                        $("phone"),
                        $("id").count().as("personCount"),
                        $("aaaa").start(),
                        $("aaaa").end(),
                        $("aaaa").proctime());

        tableEnv.toRetractStream(windowTable, Row.class).print("windowTable");
    }


    /**
     * Similar to a SQL OVER clause. Over window aggregates are computed for each row, based on a window (range) of
     * preceding and succeeding rows. See the over windows section for more details.
     * 例： 依次输入下面的内容
     * 1,11,111
     * 1,22,222
     * 1,33,333
     * 2,22,222
     * 2,33,333
     * 2,44,444
     * 输出结果
     * overWindow> (true,1,11,111,1,11,111)
     * overWindow> (true,1,22,222,2,22,111)
     * overWindow> (true,1,33,333,3,33,111)
     * overWindow> (true,2,22,222,1,22,222)
     * overWindow> (true,2,33,333,2,33,222)
     * overWindow> (true,2,44,444,3,44,222)
     */
    static void overWindow() {
        DataStream<Person> dataStream = env.socketTextStream("192.168.10.131", 10003)
                .map(i -> {
                    String[] split = i.split(",");
                    return new Person(split[0], split[1], split[2]);
                });

        Table overWindowTable = tableEnv.fromDataStream(dataStream, $("id"), $("name"), $("phone")
                , $("proctime").proctime())
                .window(
                        Over.partitionBy($("id"))
                                .orderBy($("proctime"))
                                .preceding(UNBOUNDED_RANGE)  //设置起始范围
                                .following(CURRENT_RANGE)
                                .as("windowField")
                )
                .select($("id"),
                        $("name"),
                        $("phone"),
                        $("id").count().over($("windowField")).as("personCount"),
                        $("name").max().over($("windowField")).as("nameMax"),
                        $("phone").min().over($("windowField")).as("phoneMin")


                );
        tableEnv.toRetractStream(overWindowTable,Row.class).print("overWindow");


    }
}
